package com.pap.rocketmq.config.listener;

import com.alibaba.fastjson.JSON;
import com.pap.rocketmq.entity.Mq;
import com.pap.rocketmq.repository.MqRepository;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Random;

/**
 * 针对 mq-topic 的 消息监听
 */
@Component
public class PapMqTopicTransactionListener implements TransactionListener {

    static Logger logger = LoggerFactory.getLogger(PapMqTopicTransactionListener.class);

    @Autowired
    MqRepository mqRepository;

    /**
     *
     */
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {

        LocalTransactionState lts = LocalTransactionState.UNKNOW;

        @SuppressWarnings("unchecked")
        Mq mqEntity = JSON.parseObject(new String(msg.getBody()), Mq.class);

        try {
            // TODO 此处执行 DB 操作， 并且将 事务消息 的 transactionId 进行存储，此 事务消息编号 与 业务数据进行一一匹配(后续本地消息查询使用)；

            if(new Random().nextInt(10) == 5) {
                int i = 1 / 0;
            }

            mqEntity.setId(msg.getTransactionId());
            mqRepository.save(mqEntity);

            lts = LocalTransactionState.COMMIT_MESSAGE;
        } catch (Exception e) {
            logger.error("数据库事务异常: [{}][{}]", e.getMessage(), mqEntity.toString());
            lts = LocalTransactionState.ROLLBACK_MESSAGE;
        }
        return lts;

    }

    /**
     * 去数据库查询看看是否存在已经成功发送预提交数据而没有commit成功的mq信息
     */
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        // TODO 这里进行数据判断，查询事务消息编号 对应的 业务数据是否正确存储，如果正确存储，则事务消息提交，否则进行回滚
        if (mqRepository.findById(msg.getTransactionId()) != null && mqRepository.findById(msg.getTransactionId()).isPresent()) {
            return LocalTransactionState.COMMIT_MESSAGE;
        } else {
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }

    }

}
